package com.scala.business

import com.google.gson.Gson
import com.scala.pojo.AllBasicInfoScala
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.mllib.linalg.{Matrix, Vector, Vectors}
import org.apache.spark.mllib.stat.Statistics
import org.apache.spark.rdd.RDD

/**
 * 6、统计每个会员的平均折扣率与飞行次数之间的相关性：
 * 这个指标可以帮助我们理解折扣率是否对飞行次数产生影响。
 * */
object Take06 {
  val gson: Gson = new Gson()

  def main(args: Array[String]): Unit = {
    val sparkConf = new SparkConf().setAppName("每个会员的平均折扣率与飞行次数之间的相关性").setMaster("local")
    val sc = new SparkContext(sparkConf)
    val rdd = sc.textFile("hdfs://master:9000/air_data/*")

    val res: RDD[Vector] = rdd.map((x: String) => gson.fromJson(x, classOf[AllBasicInfoScala]))
      .filter((info: AllBasicInfoScala) => info.flightCount != null && info.avgDiscount != null) //需要过滤掉空值
      .map((x: AllBasicInfoScala) => {
        val arr: Array[Double] = new Array[Double](2)
        arr(0) = x.avgDiscount // 平均折扣率
        arr(1) = x.flightCount.toDouble // 飞行次数

        Vectors.dense(arr)
      })

    System.out.println()
    // 使用 Statistics.corr() 方法计算相关性矩阵
    // Statistics.corr() 方法需要接收一个 RDD<org.apache.spark.mllib.linalg.Vector> 类型的参数
    // method: pearson,spearman
    val correlationMatrix: Matrix = Statistics.corr(res, "pearson")
    // 打印出来方便观看
    println(correlationMatrix)
    println("------------------------------------------------")

    val correlation: Double = correlationMatrix.apply(1, 0) // 获取平均折扣率和飞行次数之间的相关性值

    println("平均折扣率和飞行次数之间的相关性值为：", correlation)

    sc.stop()
  }
}
